package tianrun.ziguan.api.calculate.config.service.influxdb.impl;

import okhttp3.OkHttpClient;
import org.influxdb.InfluxDB;
import org.influxdb.InfluxDBFactory;
import org.influxdb.InfluxDBMapperException;
import org.influxdb.annotation.Column;
import org.influxdb.dto.BatchPoints;
import org.influxdb.dto.Point;
import org.influxdb.dto.Query;
import org.influxdb.dto.QueryResult;
import tianrun.ziguan.api.calculate.config.config.influxdb.InfluxDbProperties;
import tianrun.ziguan.api.calculate.config.service.influxdb.TableName;

import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLSocketFactory;
import javax.net.ssl.TrustManager;
import javax.net.ssl.X509TrustManager;
import java.lang.reflect.Field;
import java.security.cert.X509Certificate;
import java.time.Instant;
import java.time.format.DateTimeFormatter;
import java.time.format.DateTimeFormatterBuilder;
import java.time.temporal.ChronoField;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

final class QueryInfluxDb {
    private final InfluxDbProperties config;
    private final InfluxDB client;
    private final MultipleInfluxDBMapper mapper;

    public QueryInfluxDb(InfluxDbProperties config) {
        this.config = config;
        this.client = createClient();
        this.client.setDatabase(config.getDbName());
        this.client.setLogLevel(InfluxDB.LogLevel.NONE);
        this.mapper = new MultipleInfluxDBMapper(client);
    }

    public <T> List<T> query(Query query, Class<T> clz) {
        return mapper.query(query, clz);
    }
    public boolean insert(InsertValue value) {
        return mapper.insert(value);
    }
    public boolean delete(String sql,String database) {
        QueryResult queryResult = mapper.delete(sql, database);

        return !queryResult.hasError();
    }

    public <T> List<T> queryTodayData(String sql, Class<T> clz) {
        return mapper.query(new Query(sql), clz);
    }

    public InfluxDB getClient() {
        return client;
    }


    private InfluxDB createClient() {
        return InfluxDBFactory.connect(config.getUrl(), config.getUser(), config.getPassword(), build());
    }

    private OkHttpClient.Builder build() {
        return okBuilder();
    }

    private OkHttpClient.Builder okBuilder() {
        OkHttpClient.Builder target = new OkHttpClient.Builder();
        try {
            X509TrustManager trustManager = new X509TrustManager() {
                @Override
                public void checkClientTrusted(X509Certificate[] x509Certificates, String s) {
                }

                @Override
                public void checkServerTrusted(X509Certificate[] x509Certificates, String s) {
                }

                @Override
                public X509Certificate[] getAcceptedIssuers() {
                    return new X509Certificate[0];
                }
            };
            SSLContext sslContext = SSLContext.getInstance("TLS");
            sslContext.init(null, new TrustManager[]{trustManager}, null);
            SSLSocketFactory sslSocketFactory = sslContext.getSocketFactory();
            target.sslSocketFactory(sslSocketFactory, trustManager).retryOnConnectionFailure(true).hostnameVerifier((hostname, session) -> true);
            target.readTimeout(10,TimeUnit.MINUTES);
            target.writeTimeout(10,TimeUnit.MINUTES);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return target;
    }

    public boolean insertBatch(String dbName,Collection<InsertValue> values) {
        return mapper.insertBatch(dbName,values);
    }


    private static class MultipleInfluxDBMapper {

        /**
         * Data structure used to cache classes used as measurements.
         */
        private static final ConcurrentMap<String, ConcurrentMap<String, Field>> CLASS_FIELD_CACHE = new ConcurrentHashMap<>();

        private static final int FRACTION_MIN_WIDTH = 0;
        private static final int FRACTION_MAX_WIDTH = 9;
        private static final boolean ADD_DECIMAL_POINT = true;

        /**
         * When a query is executed without {@link TimeUnit}, InfluxDB returns the <code>time</code>
         * column as a RFC3339 date.
         */
        private static final DateTimeFormatter RFC3339_FORMATTER = new DateTimeFormatterBuilder()
                .appendPattern("yyyy-MM-dd'T'HH:mm:ss")
                .appendFraction(ChronoField.NANO_OF_SECOND, FRACTION_MIN_WIDTH, FRACTION_MAX_WIDTH, ADD_DECIMAL_POINT)
                .appendZoneOrOffsetId()
                .toFormatter();
        private final InfluxDB influxDB;

        public MultipleInfluxDBMapper(final InfluxDB influxDB) {
            this.influxDB = influxDB;
        }
        public QueryResult delete(String sql,String database){
            Query query = new Query(sql, database);
            return influxDB.query(query);
        }

        public <T> List<T> toPOJO(final QueryResult queryResult, final Class<T> clazz)
                throws InfluxDBMapperException {
            Objects.requireNonNull(queryResult, "queryResult");
            Objects.requireNonNull(clazz, "clazz");
            throwExceptionIfResultWithError(queryResult);
            cacheMeasurementClass(clazz);
            List<T> result = new LinkedList<>();
            List<QueryResult.Result> collect = queryResult.getResults()
                    .stream()
                    .filter(internalResult -> Objects.nonNull(internalResult) && Objects.nonNull(internalResult.getSeries()))
                    .collect(Collectors.toList());
            for (QueryResult.Result internalResult : collect) {
                for (QueryResult.Series series : internalResult.getSeries()) {
                    result.addAll(parseSeriesAs(series, clazz));
                }
            }
            return result;
        }


        void throwExceptionIfResultWithError(final QueryResult queryResult) {
            if (queryResult.getError() != null) {
                throw new InfluxDBMapperException("InfluxDB returned an error: " + queryResult.getError());
            }

            queryResult.getResults().forEach(seriesResult -> {
                if (seriesResult.getError() != null) {
                    throw new InfluxDBMapperException("InfluxDB returned an error with Series: " + seriesResult.getError());
                }
            });
        }

        void cacheMeasurementClass(final Class<?>... classVarAgrs) {
            for (Class<?> clazz : classVarAgrs) {
                if (CLASS_FIELD_CACHE.containsKey(clazz.getName())) {
                    continue;
                }
                ConcurrentMap<String, Field> influxColumnAndFieldMap = new ConcurrentHashMap<>();

                Class<?> c = clazz;
                while (c != null) {
                    for (Field field : c.getDeclaredFields()) {
                        Column colAnnotation = field.getAnnotation(Column.class);
                        if (colAnnotation != null) {
                            influxColumnAndFieldMap.put(colAnnotation.name(), field);
                        }
                    }
                    c = c.getSuperclass();
                }
                CLASS_FIELD_CACHE.putIfAbsent(clazz.getName(), influxColumnAndFieldMap);
            }
        }

        <T> List<T> parseSeriesAs(final QueryResult.Series series, final Class<T> clazz) {
            int columnSize = series.getColumns().size();
            ConcurrentMap<String, Field> colNameAndFieldMap = CLASS_FIELD_CACHE.get(clazz.getName());
            List<T> result = new ArrayList<>();
            try {
                T object = null;
                for (List<Object> row : series.getValues()) {
                    for (int i = 0; i < columnSize; i++) {
                        Field correspondingField = colNameAndFieldMap.get(series.getColumns().get(i)/*InfluxDB columnName*/);
                        if (correspondingField != null) {
                            if (object == null) {
                                object = clazz.newInstance();
                            }
                            setFieldValue(object, correspondingField, row.get(i));
                        }
                    }
                    if (series.getTags() != null && !series.getTags().isEmpty()) {
                        for (Map.Entry<String, String> entry : series.getTags().entrySet()) {
                            Field correspondingField = colNameAndFieldMap.get(entry.getKey());
                            if (correspondingField != null) {
                                setFieldValue(object, correspondingField, entry.getValue());
                            }
                        }
                    }
                    if (object instanceof TableName) {
                        ((TableName<?>) object).setTableName(series.getName());
                    }
                    if (object != null) {

                        result.add(object);
                        object = null;
                    }
                }
            } catch (InstantiationException | IllegalAccessException e) {
                throw new InfluxDBMapperException(e);
            }
            return result;
        }

        <T> void setFieldValue(final T object, final Field field, final Object value)
                throws IllegalArgumentException, IllegalAccessException {
            if (value == null) {
                return;
            }
            Class<?> fieldType = field.getType();
            try {
                if (!field.isAccessible()) {
                    field.setAccessible(true);
                }
                if (fieldValueModified(fieldType, field, object, value)
                        || fieldValueForPrimitivesModified(fieldType, field, object, value)
                        || fieldValueForPrimitiveWrappersModified(fieldType, field, object, value)) {
                    return;
                }
                String msg = "Class '%s' field '%s' is from an unsupported type '%s'.";
                throw new InfluxDBMapperException(
                        String.format(msg, object.getClass().getName(), field.getName(), field.getType()));
            } catch (ClassCastException e) {
                String msg = "Class '%s' field '%s' was defined with a different field type and caused a ClassCastException. "
                        + "The correct type is '%s' (current field value: '%s').";
                throw new InfluxDBMapperException(
                        String.format(msg, object.getClass().getName(), field.getName(), value.getClass().getName(), value));
            }
        }

        <T> boolean fieldValueModified(final Class<?> fieldType, final Field field, final T object, final Object value)
                throws IllegalArgumentException, IllegalAccessException {
            if (String.class.isAssignableFrom(fieldType)) {
                field.set(object, String.valueOf(value));
                return true;
            }
            if (Instant.class.isAssignableFrom(fieldType)) {
                Instant instant;
                if (value instanceof String) {
                    instant = Instant.from(RFC3339_FORMATTER.parse(String.valueOf(value)));
                } else if (value instanceof Long) {
                    instant = Instant.ofEpochMilli(toMillis((long) value));
                } else if (value instanceof Double) {
                    instant = Instant.ofEpochMilli(toMillis(((Double) value).longValue()));
                } else if (value instanceof Integer) {
                    instant = Instant.ofEpochMilli(toMillis(((Integer) value).longValue()));
                } else {
                    throw new InfluxDBMapperException("Unsupported type " + field.getClass() + " for field " + field.getName());
                }
                field.set(object, instant);
                return true;
            }
            return false;
        }

        <T> boolean fieldValueForPrimitivesModified(final Class<?> fieldType, final Field field, final T object,
                                                    final Object value) throws IllegalArgumentException, IllegalAccessException {
            if (double.class.isAssignableFrom(fieldType)) {
                field.setDouble(object, (Double) value);
                return true;
            }
            if (long.class.isAssignableFrom(fieldType)) {
                field.setLong(object, ((Double) value).longValue());
                return true;
            }
            if (int.class.isAssignableFrom(fieldType)) {
                field.setInt(object, ((Double) value).intValue());
                return true;
            }
            if (boolean.class.isAssignableFrom(fieldType)) {
                field.setBoolean(object, Boolean.parseBoolean(String.valueOf(value)));
                return true;
            }
            return false;
        }

        <T> boolean fieldValueForPrimitiveWrappersModified(final Class<?> fieldType, final Field field, final T object,
                                                           final Object value) throws IllegalArgumentException, IllegalAccessException {
            if (Double.class.isAssignableFrom(fieldType)) {
                field.set(object, value);
                return true;
            }
            if (Long.class.isAssignableFrom(fieldType)) {
                field.set(object, ((Double) value).longValue());
                return true;
            }
            if (Integer.class.isAssignableFrom(fieldType)) {
                field.set(object, ((Double) value).intValue());
                return true;
            }
            if (Boolean.class.isAssignableFrom(fieldType)) {
                field.set(object, Boolean.valueOf(String.valueOf(value)));
                return true;
            }
            return false;
        }

        private Long toMillis(final long value) {
            return TimeUnit.MILLISECONDS.convert(value, TimeUnit.MILLISECONDS);
        }

        public <T> List<T> query(Query query, Class<T> clazz) {
            QueryResult queryResult = influxDB.query(query);
            return toPOJO(queryResult, clazz);
        }

        public boolean insert(InsertValue value) {
            Point point = Point.measurement(value.getAssetNum())
                    .time(value.getTime(), TimeUnit.MILLISECONDS)
                    .tag("label", value.getLabel())
                    .addField("value",value.getValue())
                    .build();

            influxDB.write(point);
            return true;
        }

        public boolean insertBatch(String dbName,Collection<InsertValue> values) {
            // 批量插入数据点
            BatchPoints batchPoints = BatchPoints.database(dbName)
                    .retentionPolicy("autogen")
                    .build();

            values.forEach(value->{
                    Point point = Point.measurement(value.getAssetNum())
                    .time(value.getTime(), TimeUnit.MILLISECONDS)
                    .tag("label", value.getLabel())
                    .addField("value",value.getValue())
                    .build();
                batchPoints.point(point);

            });

            influxDB.write(batchPoints);
            //System.out.println("批量插入数据点成功！");
            return true;
        }
    }
}
